package com.atguigu.flink.chapter11;

import com.atguigu.flink.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;

public class Flink11_Time_Event {
       public static void main(String[] args) {
               Configuration configuration = new Configuration();
               //web  UI端口
               configuration.setInteger("rest.prot",10000);
               StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
               env.setParallelism(1);

               //因为加了window 后变成普通的流的，不是source？？？！！！ 不是 DataStreamSource 了
           DataStream<WaterSensor> waterSensorStream =
                   env.fromElements(new WaterSensor("sensor_1", 1000L, 10),
                           new WaterSensor("sensor_1", 2000L, 20),
                           new WaterSensor("sensor_2", 3000L, 30),
                           new WaterSensor("sensor_1", 4000L, 40),
                           new WaterSensor("sensor_1", 5000L, 50),
                           new WaterSensor("sensor_2", 6000L, 60))
                   .assignTimestampsAndWatermarks(
                           WatermarkStrategy
                                    .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                   //事件事件来自于时间戳
                                   .withTimestampAssigner((event, timestamp) -> event.getTs())
                   );


           StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
            // 1. 直接新增一个字段，用来表示  事件时间
         //  Table table = tableEnv.fromDataStream(waterSensorStream,$("id"),$("ts"),$("vc"),$("et").rowtime());

           // 2. 把现有字段，标记为事件时间
           Table table = tableEnv.fromDataStream(waterSensorStream,$("id"),$("ts").rowtime(),$("vc"));
           table.execute().print();








           }
}
